package com.webull.information.center.carwler.task;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Date;
import java.util.List;
import java.util.Locale;
import java.util.Map;
import java.util.Optional;
import java.util.TimeZone;
import java.util.concurrent.ConcurrentHashMap;
import java.util.function.Consumer;
import java.util.function.Function;

import org.apache.commons.lang3.ArrayUtils;
import org.apache.commons.lang3.StringEscapeUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.time.DateUtils;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.Logger;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import com.fasterxml.jackson.databind.JsonNode;
import com.github.stuxuhai.jpinyin.ChineseHelper;
import com.webull.framework.util.UtilDate;
import com.webull.information.center.carwler.common.json.unit.News_1_RetrieveHeadlineML_1;
import com.webull.information.center.carwler.common.model.ArchiveRel;
import com.webull.information.center.carwler.common.model.NewsInformation;
import com.webull.information.center.carwler.common.model.RtrsNewsCollect;
import com.webull.information.center.carwler.common.util.NewsTaskUtil;
import com.webull.information.center.carwler.common.util.ReutersGuideUtil;
import com.webull.information.center.carwler.dao.ArchiveDao;
import com.webull.information.center.carwler.dao.RtrsNewsCollectDao;
import com.webull.information.center.carwler.task.support.TikcerNewsProcessor;
import com.webull.information.center.common.constants.Constants;
import com.webull.information.client.model.News;
import com.webull.information.client.model.parm.NewsRelType;
import com.webull.securities.client.model.Region;
import com.webull.securities.client.model.Ticker;

/**
 * 
 * @author smj
 * @Date 2017年8月10日下午2:33:13
 * @version 1.0
 *
 */
@Component
public class News_ReutersGuide_Task2 {
    protected final Logger logger = LogManager.getLogger(getClass());

	private @Autowired ReutersGuideUtil reutersGuideUtil;
	private @Autowired RtrsNewsCollectDao rtrsNewsCollectDao;
	private @Autowired TikcerNewsProcessor tikcerNewsProcessor;
	private @Autowired @Qualifier("executor_news") ThreadPoolTaskExecutor threadPoolTaskExecutor;
	private @Autowired ArchiveDao archiveDao;
	private final int rtrsMaxCount = 50; // 路透接口每次获取的最大数量，用来控制包大小，以及分页判定逻辑

    /**
	 * 
	 * @param region
	 * @return
	 */
	public Integer fullReutersGuideStockNews(Region region) {
		logger.info("News_ReutersGuide_Task2 fullReutersGuideStockNews region:{},code:{}", region.getId(),
				region.getAlias());
		RtrsNewsCollect newsCollect = rtrsNewsCollectDao.queryCollect(region.getId());

		Date pointSt = getPointSt(newsCollect);// 本次需执行的时间切片(最少也执行5小时，保证动态的完整性)
		Date pointEd = new Date();

		logger.info("News_ReutersGuide_Task2 fullReutersGuideStockNews end region:{},pointSt:{},pointEd:{}",
				region.getId(), pointSt, pointEd);
		Periods periods = producePeriods(pointSt, pointEd);
		int runTask = runRegionTask(region, periods, newsCollect.getId());

		logger.info("News_ReutersGuide_Task2 fullReutersGuideStockNews end region:{},code:{}",
				region.getId(), region.getAlias(), runTask);
		return runTask;
    }

	private static Date getPointSt(RtrsNewsCollect newsCollect) {
		Date pointSt = DateUtils.addHours(new Date(), -5);
		Date oldExecTime = newsCollect.getOldExecTime();
		Date execTime = newsCollect.getExecTime();
		if (oldExecTime != null && oldExecTime.before(execTime)) {
			execTime = oldExecTime;
		}
		pointSt = execTime.before(pointSt) ? execTime : pointSt;// 本次需执行的时间切片(最少也执行5小时，保证动态的完整性)
		return pointSt;
	}

	private int runRegionTask(Region region, Periods periods, Integer collectId) {
		if (periods.getPositive() != null && !periods.getPositive().isEmpty()) {
			// 正向更新
			runRegionTask1(region, periods.getPositive(), collectId, false);
		}
		if (periods.getReverse() != null && !periods.getReverse().isEmpty()) {
			// 异步执行补偿过程
			new Thread(() -> runRegionTask1(region, periods.getReverse(), collectId, true)).start();
		}
		return 1;
	}

	private void runRegionTask1(Region region, List<Date[]> slices, Integer collectId, boolean reverse) {
		/*
		 * 处理单个时间切片的Consumer
		 */
		final Consumer<Date[]> consumerRegion = new Consumer<Date[]>() {

			@Override
			public void accept(Date[] sli0) {
				String regionCode = region.getAlias();

				logger.info("News_ReutersGuide_Task2 runRegionTask1.consumerRegion.accept region:{},code:{},sli0:{}",
						region.getId(), region.getAlias(), ArrayUtils.toString(sli0));

				JsonNode rtrsResult0 = processRtrsResult0(regionCode, sli0);
				JsonNode newsItems = Optional.ofNullable(rtrsResult0)
						.map(v -> v.at("/RetrieveHeadlineML_Response_1/HeadlineMLResponse/HEADLINEML/HL"))
						.filter(v -> v.isArray()).orElse(null);
						/*
				 * 解析条目
				 */
				if (newsItems == null || newsItems.size() <= 0)
					return;

				for (JsonNode newsItem : newsItems) {
					processReutersNews(newsItem, region.getId());
				}

				if (newsItems.size() >= rtrsMaxCount) {
					/*
					 * 没有取完，且取到的最近时间是。。。
					 */
					// "2017-08-10T06:12:27+00:00"
					Date endD1 = Optional
							.ofNullable(rtrsResult0
									.at("/RetrieveHeadlineML_Response_1/HeadlineMLResponse/Context/Older").asText(null))
							.map(vdate -> UtilDate.parse(vdate, "yyyy-MM-dd'T'HH:mm:ssXXX")).orElse(null);

					if (endD1 != null) {
						Date[] sli1 = new Date[] { sli0[0], endD1 };
						logger.info(
								"News_ReutersGuide_Task2 runRegionTask1.consumerRegion.accept region:{},code:{},sli1:{}",
								region.getId(), region.getAlias(), ArrayUtils.toString(sli1));

						accept(sli1);
					}
				}

			}
		};


		 
		if (reverse) { // 补偿执行，切片过多，应分批处理
			List<Date[]> brief = new ArrayList<>();
			for (int i = 0; i < slices.size(); i++) {
				brief.add(slices.get(i));
				if (brief.size() > 5 || i >= slices.size() - 1) {
					int errors = NewsTaskUtil.multiThreadProcess(brief, consumerRegion, threadPoolTaskExecutor);
					if (errors <= 0) {
						RtrsNewsCollect toUpdate = RtrsNewsCollect.produceById(collectId);
						toUpdate.setOldExecTime(slices.get(i)[1]);
						rtrsNewsCollectDao.update(toUpdate);// 记录每个批次执行完后的补偿point
						/*
						 * 初始化
						 */
						brief = new ArrayList<>();
					} else {
						logger.warn(
								"News_ReutersGuide_Task2 runRegionTask1.reverse.error region:{},code:{},sli1:{},errors:{}",
								region.getId(), region.getAlias(), ArrayUtils.toString(brief.get(0)), errors);
						return;
					}
				}
			}

			RtrsNewsCollect toUpdate = RtrsNewsCollect.produceById(collectId);
			toUpdate.setExecCount(1);
			rtrsNewsCollectDao.update(toUpdate);

			rtrsNewsCollectDao.completeReverse(collectId);// 补偿完成后，清空补偿point
		} else { // 一次性处理执行
			int errors = NewsTaskUtil.multiThreadProcess(slices, consumerRegion, threadPoolTaskExecutor);
			if (errors <= 0) {
				RtrsNewsCollect toUpdate = RtrsNewsCollect.produceById(collectId);
				toUpdate.setExecTime(slices.get(slices.size() - 1)[1]);
				toUpdate.setExecStat(2);
				toUpdate.setExecCount(1);
				rtrsNewsCollectDao.update(toUpdate);
			} else {
				logger.warn("News_ReutersGuide_Task2 runRegionTask1.!reverse.error region:{},code:{},sli1:{},errors:{}",
						region.getId(), region.getAlias(), ArrayUtils.toString(slices.get(0)), errors);
			}
		}
	}

	private JsonNode processRtrsResult0(String regionCode, Date[] slice0) {
		String jsonParam2 = News_1_RetrieveHeadlineML_1.getRegionLangsRequestJson(regionCode, rtrsMaxCount, slice0[0],
				slice0[1]);
		JsonNode jsonNode2 = reutersGuideUtil.getReutersData(News_1_RetrieveHeadlineML_1.restUrl, jsonParam2);
		return jsonNode2;
	}

	/**
	 * 根据新闻查找新闻所对应的company或者ticker
	 */
	private void processReutersNews(JsonNode newsItem, Integer regionId) {

		String[] companys = StringUtils.split(newsItem.at("/CO").asText(), " ");
		if (companys == null || companys.length == 0)
			return;

		List<Ticker> tickers = new ArrayList<>();
		for (String nRic:companys) {
			List<Ticker> apply = archiveCache.apply(nRic);
			if (apply != null && !apply.isEmpty()) {
				tickers.addAll(apply);
			}
		}

		if (tickers.isEmpty())
			return;

		NewsInformation info = proudceNewsInfo(newsItem, regionId);
		if (!checkNewsInfo(info))
			return;

		if (Constants.lang_zh_hant.equalsIgnoreCase(info.getLanguage())) {// 繁体翻译简体
			NewsInformation info2 = new NewsInformation();
			BeanUtils.copyProperties(info, info2);
			if (info2.getTitle() != null)
				info2.setTitle(ChineseHelper.convertToSimplifiedChinese(info2.getTitle()));
			if (info2.getSourceName() != null)
				info2.setSourceName(ChineseHelper.convertToSimplifiedChinese(info2.getSourceName()));

			info2.setLanguage(Constants.lang_zh);
			info2.setUrl(info2.getUrl() + "_simplified");

			tickers.forEach(
					tk -> tikcerNewsProcessor.bindTargetNewsRecord(info2, tickers.get(0).getId(), NewsRelType.ticker));
		}

		tickers.forEach(
				tk -> tikcerNewsProcessor.bindTargetNewsRecord(info, tickers.get(0).getId(), NewsRelType.ticker));
	}

	private NewsInformation proudceNewsInfo(JsonNode hl0, Integer regionId) {

		NewsInformation info = new NewsInformation();

		String urn = hl0.at("/ID").asText();
		String source = hl0.at("/PR").asText();
		String lang = hl0.at("/LN").asText();

		String title = hl0.at("/HT").asText();
		// System.out.println("********--->" + title);

		if (StringUtils.isNotBlank(title)) {
			title = StringUtils.removePattern(title, "(?i)<Origin [^<]{1,}>[^<]{1,}</Origin>");
			title = StringUtils.remove(title, "(with URL)");
			title = StringEscapeUtils.unescapeXml(title);// 转义特殊字符
			title = StringEscapeUtils.unescapeHtml4(title);// 转义特殊字符
		}

		try {
			Date newsTime = UtilDate.parse(hl0.at("/CT").asText(), Locale.US, TimeZone.getTimeZone("GMT"),
					"yyyy-MM-dd'T'HH:mm:ssXXX");
			info.setNewsTime(newsTime);// // 2016-10-23T23:28:05+00:00

		} catch (Exception e) {
			logger.error("proudceNewsInfo StockReuteGuideHandler error:{}", e);
		}

		info.setUrl(StringUtils.stripToNull(urn));
		info.setTitle(StringUtils.stripToNull(title));

		if (Constants.lang_zh.equalsIgnoreCase(lang)) {
//			char[] a = title.toCharArray();
//			lang: for (int i = 0; a != null && i < a.length; i++) {
//				if (ChineseHelper.isTraditionalChinese(a[i])) {
			lang = Constants.lang_zh_hant;
//					break lang;
//				}
//			}
		}

		info.setLanguage(StringUtils.lowerCase(lang));// zh/en
		info.setSourceName(StringUtils.stripToNull(source));
		info.setCollectSource(News.CollectSource.reutersGuide);
		info.setRegionId(regionId);
		info.setSiteType(News.SiteType.NORMAL);
		return info;
	}

	private static boolean checkNewsInfo(NewsInformation info) {
		if (info == null)
			return false;
		if (info.getNewsTime() == null)
			return false;
		if (StringUtils.isBlank(info.getTitle()))
			return false;
		if (StringUtils.isBlank(info.getLanguage()))
			return false;
		return true;
	}


	/*
	 * 计算正常采集和补偿采集的时间切片集合
	 * 
	 * @param start
	 * 
	 * @param end
	 * 
	 * @return
	 */
	private static Periods producePeriods(final Date start, final Date end) {
		final Periods perd = new Periods();
		final Date critical = DateUtils.addDays(end, -1);
		Date start0 = end;

		for0: for (;;) {
			Date end0 = start0;
			start0 = DateUtils.addHours(start0, -1);
			if (end0.after(start)) {
				if (end0.after(critical))
					perd.getPositive().add(new Date[] { start0, end0 });
				else
					perd.getReverse().add(new Date[] { start0, end0 });
			} else
				break for0;
		}
		Collections.sort(perd.getPositive(), (c1, c2) -> c2[0].compareTo(c1[0]));// 调整顺序
		Collections.sort(perd.getReverse(), (c1, c2) -> c1[0].compareTo(c2[0]));// 调整顺序
		return perd;
	}

	public static void main(String[] args) {
		// Periods periods = producePeriods(DateUtils.addHours(new Date(), -24), new
		// Date());

		// System.out.println(ArrayUtils.toString(periods.getReverse()));
		Date dd = UtilDate.parse("2017-08-10T06:12:27+00:00", "yyyy-MM-dd'T'HH:mm:ssXXX");
		System.out.println(UtilDate.formatTZ(dd, UtilDate.getTimeZone(7)));
		Date dd1 = UtilDate.parse("2017-08-10T06:12:27+08:00", "yyyy-MM-dd'T'HH:mm:ssXXX");
		System.out.println(UtilDate.formatTZ(dd1, UtilDate.getTimeZone(8)));
	}

	/**
	 * 时间切片集合
	 * 
	 * @author smj
	 * @Date 2017年8月10日下午8:15:42
	 * @version 1.0
	 *
	 */
	private static class Periods {
		private List<Date[]> positive = new ArrayList<>(); // 本次正常更新的时间片
		private List<Date[]> reverse = new ArrayList<>();// 本次补偿更新时间片

		private List<Date[]> getPositive() {
			return positive;
		}


		private List<Date[]> getReverse() {
			return reverse;
		}
	}

	/**
	 * 档案缓存器
	 */
	private Function<String, List<Ticker>> archiveCache = new Function<String, List<Ticker>>() {

		private Date created = new Date();
		private Map<String, List<Ticker>> tickers = new ConcurrentHashMap<>();
		
		@Override
		public List<Ticker> apply(String nRic) {
			if (DateUtils.addHours(new Date(), -8).after(created)) {// refresh
				tickers = new ConcurrentHashMap<>();
				created = new Date();
				logger.info("News_ReutersGuide_Task2.archiveCache refresh....");
			}
			if (tickers.get(nRic) == null) {// init item

				List<ArchiveRel> srcIds = archiveDao.queryArchivesBySrcId(nRic);
				if (srcIds != null && !srcIds.isEmpty()) {
					List<Ticker> tickes0 = new ArrayList<>();
					srcIds.forEach(v -> {
						Ticker tt = new Ticker();
						tt.setId(v.getWlId());
						tt.setTickerType(v.getType());
						tickes0.add(tt);
					});
					tickers.put(nRic, tickes0);
				} else {
					tickers.put(nRic, Empty);
				}

			}
			// get item
			return tickers.getOrDefault(nRic, Empty);
		}

		private List<Ticker> Empty = new ArrayList<>();
	};

}
